package com.juneyao.bigdata.config;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.data.hadoop.hbase.HbaseTemplate;

import java.io.IOException;

/**
 * @author quanbiao
 * @date 2018-02-06
 * Hbase 连接池
 */
@org.springframework.context.annotation.Configuration
@EnableConfigurationProperties(HbaseProperties.class)
@ConditionalOnClass(HbaseTemplate.class)
public class HbaseAutoConfiguration {

    /**
     * zk连接地址
     */
    private static final String HBASE_QUORUM = "hbase.zookeeper.quorum";

    /**
     * zk端口号
     */
    private static final String HBASE_ZK_PORT = "hbase.zookeeper.property.clientPort";

    /**
     * hbase连接池个数，默认是1，连接regonServer个数
     */
    private static final String HBASE_CLIENT_IPC_POOL_SIZE= "hbase.client.ipc.pool.size";

    @Autowired
    private HbaseProperties hbaseProperties;

    @Bean
    @ConditionalOnMissingBean(HbaseTemplate.class)
    public HbaseTemplate hbaseTemplate() {
        Configuration configuration = getConfiguration();
        return new HbaseTemplate(configuration);
    }

    @Bean
    @ConditionalOnMissingBean(Connection.class)
    public Connection connection() {
        Configuration configuration = getConfiguration();
        try {
            return ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取hbase configuration
     * @return
     */
    private Configuration getConfiguration(){
        Configuration configuration = HBaseConfiguration.create();
        configuration.set(HBASE_QUORUM, this.hbaseProperties.getQuorum());
        configuration.set(HBASE_ZK_PORT, this.hbaseProperties.getZkPort());
        configuration.set(HBASE_CLIENT_IPC_POOL_SIZE, "3");
        return configuration;
    }
}
